package window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Create by zhh on 2022/9/13.
 */
public class WindowAPP {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //test01(env);
        //test02(env);
        //test03(env);
        //test04(env);
        test05(env);
        env.execute("windowApp");
    }
    public static void test01(StreamExecutionEnvironment env){
        env.socketTextStream("192.168.32.222", 9527)
                //no-Keyed
//                .map(new MapFunction<String, Integer>() {
//                    @Override
//                    public Integer map(String s) throws Exception {
//                        return Integer.parseInt(s);
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum(0)
//                .print();
                //Keyed
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String s) throws Exception {
                        String[] splits = s.split(",");
                        return Tuple2.of(splits[0].trim(),Integer.parseInt(splits[1].trim()));
                    }
                })
                .keyBy(x->x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();

    }

    public static void test02(StreamExecutionEnvironment env){
        env.socketTextStream("192.168.32.222", 9527)
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String s) throws Exception {
                        String[] splits = s.split(",");
                        return Tuple2.of(splits[0].trim(),Integer.parseInt(splits[1].trim()));
                    }
                })
                .keyBy(x->x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> v1, Tuple2<String, Integer> v2) throws Exception {
                        System.out.println("v1 = " + v1 + ", v2 = " + v2);
                        return Tuple2.of(v1.f0,v1.f1+v2.f1);
                    }
                })
                .print();

    }

    public static void test03(StreamExecutionEnvironment env){
        env.socketTextStream("192.168.32.222", 9527)
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String s) throws Exception {
                        return Tuple2.of("zh",Integer.parseInt(s));
                    }
                })
                .keyBy(x->x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new ZHProcessWindowFunction())
                .print();
    }

    //基于ProcessingTime的Non-Keyed的会话窗口编程测试
    public static void test04(StreamExecutionEnvironment env){
        env.socketTextStream("192.168.32.222", 9527)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String s) throws Exception {
                        return Integer.parseInt(s);
                    }
                })
                .windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))//设置了固定间隔5秒的 processing-time session 窗口
                .sum(0)
                .print();
    }

    //基于ProcessingTime的Non-Keyed和Keyed的滑动窗口编程测试
    public static void test05(StreamExecutionEnvironment env){
        //Non-Keyed
        env.socketTextStream("192.168.32.222", 9527)
//                .map(new MapFunction<String, Integer>() {
//                    @Override
//                    public Integer map(String s) throws Exception {
//                        return Integer.parseInt(s);
//                    }
//                })
//                .windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(3)))//设置了窗口大小5秒，滑动3秒的processing-time 窗口
//                .sum(0)
//                .print();

        //Keyed
                .map(new MapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String s) throws Exception {
                        String[] splits = s.split(",");
                        return Tuple2.of(splits[0].trim(),Integer.parseInt(splits[1].trim()));
                    }
                })
                .keyBy(x->x.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(3)))
                .sum(1)
                .print();
    }
}

